﻿using System;
using System.Collections.Generic;
using System.Text;
using Newtonsoft.Json;
using RabbitMQ.Client.MessagePatterns;

namespace smp.RabbitMq
{
    internal class Publisher : RabbitMqBusBase, IPublisher
    {
        internal Publisher(IRabbitMqConfiguration configuration) 
            : base(configuration)
        {
        }

        ~Publisher()
        {
            Dispose(false);
        }

        #region Implementation of IDisposable

        public void Dispose()
        {
            Dispose(true);
            GC.SuppressFinalize(this);
        }

        #endregion

        #region Implementation of IPublisher

        public void Publish<TMessage>(TMessage message)
        {
            Publish(message, 0);
        }

        public void Publish(Type messageType, object message)
        {
            Publish(messageType, message, 0);
        }

        public void Publish<TMessage>(TMessage message, byte priority)
        {
            Publish(typeof (TMessage).AssemblyQualifiedName, SerializeMessage(message), priority);
        }

        public void Publish(Type messageType, object message, byte priority)
        {
            Publish(messageType.AssemblyQualifiedName, SerializeMessage(message), priority);
        }

        public void Publish(string assemblyQualifiedMessageType, string message)
        {
            Publish(assemblyQualifiedMessageType, message, 0);
        }

        public void Publish(string assemblyQualifiedMessageType, string message, byte priority)
        {
            PublishToExchange(new Envelope(assemblyQualifiedMessageType, message), Configuration.Exchange, Configuration.RoutingKey, priority);
        }



        public void FuturePublish<TMessage>(TMessage message, int sleepTimeInMilliSeconds)
        {
            FuturePublish(typeof (TMessage).AssemblyQualifiedName, SerializeMessage(message), sleepTimeInMilliSeconds, 0);
        }

        public void FuturePublish<TMessage>(TMessage message, TimeSpan sleepTimeSpan)
        {
            FuturePublish(typeof(TMessage).AssemblyQualifiedName, SerializeMessage(message), (int)sleepTimeSpan.TotalMilliseconds, 0);
        }

        public void FuturePublish<TMessage>(TMessage message, DateTime futurePublishDateTime)
        {
            FuturePublish(typeof (TMessage).AssemblyQualifiedName, SerializeMessage(message), futurePublishDateTime, 0);
        }

        public void FuturePublish<TMessage>(TMessage message, int sleepTimeInMilliSeconds, byte priority)
        {
            FuturePublish(typeof(TMessage).AssemblyQualifiedName, SerializeMessage(message), sleepTimeInMilliSeconds, priority);
        }

        public void FuturePublish<TMessage>(TMessage message, TimeSpan sleepTimeSpan, byte priority)
        {
            FuturePublish(typeof(TMessage).AssemblyQualifiedName, SerializeMessage(message), (int)sleepTimeSpan.TotalMilliseconds, priority);
        }

        public void FuturePublish<TMessage>(TMessage message, DateTime futurePublishDateTime, byte priority)
        {
            FuturePublish(typeof(TMessage).AssemblyQualifiedName, SerializeMessage(message), futurePublishDateTime, priority);
        }

        public void FuturePublish(Type messageType, object message, int sleepTimeInMilliSeconds)
        {
            FuturePublish(messageType.AssemblyQualifiedName, SerializeMessage(message), sleepTimeInMilliSeconds, 0);
        }

        public void FuturePublish(Type messageType, object message, TimeSpan sleepTimeSpan)
        {
            FuturePublish(messageType.AssemblyQualifiedName, SerializeMessage(message), (int)sleepTimeSpan.TotalMilliseconds, 0);
        }

        public void FuturePublish(Type messageType, object message, DateTime futurePublishDateTime)
        {
            FuturePublish(messageType.AssemblyQualifiedName, SerializeMessage(message), futurePublishDateTime, 0);
        }

        public void FuturePublish(Type messageType, object message, int sleepTimeInMilliSeconds, byte priority)
        {
            FuturePublish(messageType.AssemblyQualifiedName, SerializeMessage(message), sleepTimeInMilliSeconds, priority);
        }

        public void FuturePublish(Type messageType, object message, TimeSpan sleepTimeSpan, byte priority)
        {
            FuturePublish(messageType.AssemblyQualifiedName, SerializeMessage(message), (int)sleepTimeSpan.TotalMilliseconds, priority);
        }

        public void FuturePublish(Type messageType, object message, DateTime futurePublishDateTime, byte priority)
        {
            FuturePublish(messageType.AssemblyQualifiedName, SerializeMessage(message), futurePublishDateTime, priority);
        }

        public void FuturePublish(string assemblyQualifiedMessageType, string message, int sleepTimeInMilliSeconds)
        {
            FuturePublish(assemblyQualifiedMessageType, message, sleepTimeInMilliSeconds, 0);
        }

        public void FuturePublish(string assemblyQualifiedMessageType, string message, TimeSpan sleepTimeSpan)
        {
            FuturePublish(assemblyQualifiedMessageType, message, (int)sleepTimeSpan.TotalMilliseconds, 0);
        }

        public void FuturePublish(string assemblyQualifiedMessageType, string message, DateTime futurePublishDateTime)
        {
            FuturePublish(assemblyQualifiedMessageType, message, futurePublishDateTime, 0);
        }

        public void FuturePublish(string assemblyQualifiedMessageType, string message, int sleepTimeInMilliSeconds, byte priority)
        {
            var sleepExchangeName = CreateSleepQueueReturningExchange(sleepTimeInMilliSeconds);
            PublishToExchange(new Envelope(assemblyQualifiedMessageType, message), sleepExchangeName, Configuration.RoutingKey, priority);
        }

        public void FuturePublish(string assemblyQualifiedMessageType, string message, TimeSpan sleepTimeSpan, byte priority)
        {
            FuturePublish(assemblyQualifiedMessageType, message, (int) sleepTimeSpan.TotalMilliseconds, priority);
        }

        public void FuturePublish(string assemblyQualifiedMessageType, string message, DateTime futurePublishDateTime, byte priority)
        {
            var now = DateTime.Now;
            if (futurePublishDateTime <= now)
            {
                Publish(message);
                return;
            }
            var waitTime = (int)(futurePublishDateTime - now).TotalMilliseconds;
            FuturePublish(assemblyQualifiedMessageType, message, waitTime, priority);
        }

        #endregion

        #region private methods

        private string CreateSleepQueueReturningExchange(int sleepTimeInMilliSeconds)
        {
            var sleepExchange = Configuration.Exchange + "_SLEEP_" + sleepTimeInMilliSeconds;
            Channel.ExchangeDeclare(sleepExchange, Configuration.ExchangeType.ToString(), Configuration.DurableExchange,
                                    Configuration.AutoDeleteExchange, null);
            Log.Debug("Declared sleep exchange {0} for target exchange {1}", sleepExchange, Configuration.Exchange);

            var sleepQueue = Configuration.Exchange + "_SLEEPQUEUE_" + sleepTimeInMilliSeconds;
            var queueArgs = new Dictionary<string, object>
                                {
                                    {"x-dead-letter-exchange", Configuration.Exchange},
                                    {"x-message-ttl", sleepTimeInMilliSeconds},
                                    {"x-expires", 3600000}
                                };
            Channel.QueueDeclare(sleepQueue, Configuration.DurableQueue, false, Configuration.AutoDeleteQueue, queueArgs);
            Log.Debug("Declared sleep Queue {0} for target exhange {1}", sleepQueue, Configuration.Exchange);

            Channel.QueueBind(sleepQueue, sleepExchange, Configuration.RoutingKey, null);
            Log.Debug("Bound sleep queue {0} to sleep exchange {1} on routing key {2}", sleepQueue, sleepExchange, Configuration.RoutingKey);

            return sleepExchange;
        }

        internal void PublishToExchange(Envelope envelope, string exchangeName, string routingKey, byte priority)
        {
            var serializedMessage = JsonConvert.SerializeObject(envelope);
            var packet = Encoding.ASCII.GetBytes(serializedMessage);
            try
            {
                var properties = Channel.CreateBasicProperties();
                properties.SetPersistent(Configuration.DurableExchange);
                properties.Priority = priority;
                Channel.BasicPublish(exchangeName, routingKey, properties, packet);
                Log.Info("Published message : {1} to exchange : {0}", exchangeName, serializedMessage);
            }
            catch (Exception ex)
            {
                var error = new Exception(string.Format("An error was throws when trying to send message {0} on exchange {1}", serializedMessage, exchangeName), ex);
                Log.Error(error);
                throw error;
            }
        }

        #endregion
    }
}
